/*
 * Copyright (c) 2011 Imre Fazekas.
 * All rights reserved.
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * Redistributions of source code must retain the above copyright notice, this
 * list of conditions and the following disclaimer.
 *
 * Redistributions in binary form must reproduce the above copyright notice,
 * this list of conditions and the following disclaimer in the documentation
 * and/or other materials provided with the distribution.
 * Neither the name of the Brillien nor the names of its
 * terms and concepts may be used to endorse or promote products derived from this
 * software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

package com.vii.brillien.core.component;

import com.vii.brillien.core.management.BrillienError;
import com.vii.brillien.core.management.BrillienServices;
import com.vii.brillien.core.management.FlowServices;
import com.vii.brillien.core.management.db.TransactionServices;
import com.vii.brillien.ignition.transport.BrillienCommunication;
import com.vii.brillien.kernel.BrillienException;
import com.vii.brillien.kernel.annotations.Activate;
import com.vii.brillien.kernel.annotations.NullResponseAllowed;
import com.vii.brillien.kernel.annotations.PresenceService;
import com.vii.brillien.kernel.annotations.lifecycle.Consonant;
import com.vii.brillien.kernel.axiom.atomic.Flow;
import com.vii.brillien.kernel.axiom.atomic.Presence;
import com.vii.brillien.kernel.axiom.component.State;
import com.vii.brillien.kernel.axiom.transport.Communication;
import com.vii.streamline.services.ThreadServices;
import com.vii.streamline.services.db.transaction.FXID;
import com.vii.streamline.services.error.StreamLineException;
import com.vii.streamline.services.reflection.ReflectionServices;
import com.vii.streamline.structures.collections.InnerList;

import java.lang.reflect.Method;
import java.sql.Connection;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.TimeUnit;

import static com.vii.streamline.services.reflection.ReflectionServices.invokeMethodALike;

/**
 * Specialized SuperPresence type to provide Flow implementation
 */
@PresenceService( logLevel = PresenceService.FINE )
@Consonant( flowIDPrefix = "SuperFlow" )
public class SuperFlow <P extends Presence, R> extends AbstractContext<P,R> implements Flow<P,BrillienCommunication,R> {

    /**
     * Unique ID of the Flow
     */
    protected String                    flowID;

    /**
     * Globally unique Xid of the Flow
     */
    protected FXID                      fXid;

    /**
     * Tells whether this Flow represents an action which cannot fail, so cluster replication is needed.
     */
    protected boolean                   indispensable;

    /**
     * Tells whether this flow has been suspended.
     */
    protected boolean                   suspended;

    /**
     * Result object of the flow.
     */
    protected Object                    result;

    /**
     * Data received when the flow has been activated
     */
    protected BrillienCommunication     activationData;

    /**
     * Counts the number of pending asynchronous communication started by Presences in this flow structure
     */
    protected int                       pendingAsyncCommunications;

    /**
     * Constructor
     */
    public SuperFlow( ) {
    }

    /**
     * Constructor
     */
    public SuperFlow(String flowID) {
        this( flowID, null );
    }

    /**
     * Constructor
     */
    protected SuperFlow(String flowID, Flow parentFlow) {
        this.flowID         = flowID;
        try {
            this.setFlow( parentFlow );
        } catch (BrillienException e) {
            exceptionLog( "Superflow", "constructor", e );
        }
    }

    @Override
    public void setFlow(Flow flow) throws BrillienException {
        if( flow != null){
            flow.addSubUnits( this );
            this.flow = flow;
        }
    }

    @Override
    public boolean isIndispensable() {
        return indispensable;
    }

    public void setIndispensable(boolean indispensable) {
        this.indispensable = indispensable;
    }

    @Override
    public String getFlowID() {
        return flowID;
    }

    @Override
    public Flow<P,BrillienCommunication,R> setFlowID(String flowID) {
        this.flowID = flowID;

        return this;
    }

    /**
     * Getter method for field fxid;
     */
    public FXID getFXid() {
        return fXid;
    }

    /**
     * Setter method for field fxid;
     */
    public void setFXid(FXID fXid) {
        this.fXid = fXid;
    }                             

    @Override
    protected Object invokeRequestProcessor( BrillienCommunication message ) throws BrillienException {
        try {
            String msg = getSubject(message);
            Method m = ReflectionServices.getMethod( this.getClass(), msg );
            Object result = null;

            if( m != null ){
                authorizeCall( message.getSender(), message.getSession(), m );
                
                if( m.getAnnotation( Activate.class ) != null ){
                    Activate act = m.getAnnotation( Activate.class );
                    Method am = ReflectionServices.getMethod( this.getClass(), "activate" );
                    invokeMethodALike( //invokeMethod(
                        "activate", this, message, 
                        (Long)( act.timeout() != Activate.DEFAULT_TIMEOUT ? act.timeout() : Activate.DEFAULT_TIMEOUT )
                    );
                    if( act.timeout() != Activate.DEFAULT_TIMEOUT ){
                        ThreadServices.getStpe().schedule( new Runnable(){
                            public void run() {
                                if( !flow.isUnitInExitState() || flow.isUnitWaitingForResponse() )
                                    aspirationTimeIsUp();
                            }
                        }, act.timeout(), TimeUnit.MILLISECONDS );
                    }
                }
                String cause = null;
                try{
                    Object[] parameters = null;
                    try{
                        parameters = message.acquireParameters( m );
                    } catch( Throwable t ){ throw new BrillienException( "Cannot map parameters", t ); }

                    validateMessage( parameters );
                    result = invokeMethodALike( m, this, parameters ); //invokeMethod( m, this, parameters );
                } catch(Throwable t){ cause = t.getMessage(); throw t; }
                finally{
                    if( m.getAnnotation( Activate.class ) != null && !this.isWaitingForResponse() && !this.suspended ){
                        if( cause != null )
                            oppress( cause );
                        else
                            terminate();
                    }
                }
            }
            else{
                m = ReflectionServices.getMethod( this.getClass(), "processMessage" );

                authorizeCall( message.getSender(), message.getSession(), m );
                
                result = invokeMethodALike( m, this, new InnerList( msg, //invokeMethod( m, this, new InnerList( msg,
                    message.getParameters()
                ).asArray() );
            }

            return  result != null
                    ? result
                    : ( m.getAnnotation( NullResponseAllowed.class ) != null
                        ? BrillienServices.NULL_ANSWER
                        : null
                      );
        } catch (BrillienException be) {
            throw be;
        } catch (Throwable t) {
            throw new BrillienException( StreamLineException.ERROR_SERVER_ERROR, t );
        }
    }

    @Override
    protected Object invokeResultProcessor( BrillienCommunication message ) throws BrillienException {
        try {
            String msg = getSubject(message);
            Method m = ReflectionServices.getMethod(this.getClass(), msg);


            if( message.getOriginalType() == Communication.TYPE_AGET ){
                report( BrillienServices.REPORT_CLOSED_ASYNC_COMMUNICATION );
            }

            Object o = null;
            if( m != null ){
                authorizeCall( message.getSender(), message.getSession(), m );
                
                Object response = null;
                try{
                    response = message.acquireResponse( m.getParameterTypes()[0] );
                } catch( Throwable t ){ throw new BrillienException( "Cannot map parameters", t ); }

                validateMessage( response );

                o = invokeMethodALike( m, this, response ); //invokeMethod( m, this, response );
            }
            else{
                m = ReflectionServices.getMethod( this.getClass(), "processMessage" );

                authorizeCall( message.getSender(), message.getSession(), m );
                
                o = invokeMethodALike( m, this, //invokeMethod( m, this,
                    new InnerList( msg,
                        new Object[]{ message.getResponse() }
                    ).asArray()
                );
            }
            
            if( !this.isWaitingForResponse() && !this.suspended ){
                terminate();
            }

            return o;
        } catch (BrillienException be) {
            throw be;
        } catch (Throwable t) {
            throw new BrillienException( StreamLineException.ERROR_SERVER_ERROR, t );
        }
    }

    @Override
    public Object errorReceived( BrillienCommunication message ) throws BrillienException{
        exceptionLog( this.getClass().getName(), "errorReceived", new BrillienException( message.<String>getResponse() ) );

        if( message.getOriginalType() == Communication.TYPE_AGET )
            report( this, BrillienServices.REPORT_CLOSED_ASYNC_COMMUNICATION );

        return new BrillienError( message.getErrorCode(), message.<String>getResponse(), message.getErrorValue() );
    }
    
    /**
     * Creates a hashmap storing the actual states of the given flow 
     */
    protected HashMap<String, String> defineStateSpace( ){
        HashMap<String, String> space = new HashMap<String, String>();

        for( State state : getActualStates() )
            space.put( state.getName(), state.getClass().getName() );

        return space;
    }

    /**
     * Restores the actual states of the given flow. Called while the flow is beeing reactivated.
     */
    protected void restoreStateSpace( HashMap<String, String> stateSpace ) throws BrillienException {
        for( String name : stateSpace.keySet() ){
            try {
                State s = (State) this.getClass().getClassLoader().loadClass( stateSpace.get(name) ).newInstance();
                s.setName( name );
                addState( s );
            } catch (Exception e) {
                throw new BrillienException( e.getMessage(), e );
            }
        }
    }

    /**
     * This method is to addExtension, customize the state space of a flow. To be overriden in the child classes if needed.
     */
    protected void expandStateSpace( HashMap<String, Object> space ){

    }

    /**
     * This method is to recollect the custom state space of a flow. To be overriden in the child classes if needed.
     */
    protected void recollectStateSpace( HashMap<String, Object> space ){

    }

    @Override
    public HashMap<String, Object> getStateSpace(){
        HashMap<String, Object> space = new HashMap<String, Object>();

        expandStateSpace( space );

        space.put("pendingAsyncCommunications", pendingAsyncCommunications );
        space.put("states", defineStateSpace() );

        return space;
    }

    @Override
    public void setStateSpace( HashMap<String, Object> stateSpace ) throws BrillienException {
        this.pendingAsyncCommunications =
                stateSpace.get("pendingAsyncCommunications") instanceof Long
                ? ((Long)stateSpace.get("pendingAsyncCommunications")).intValue()
                : (Integer)stateSpace.get("pendingAsyncCommunications");
        restoreStateSpace( (HashMap<String, String>) stateSpace.get("states") );
        recollectStateSpace( stateSpace );
    }

    public boolean equals(Flow flow){
        return flowID != null && flow != null && flow.getFlowID() != null && flowID.equals( flow.getFlowID() );
    }

    @Override
    public boolean equals(Object o){
        return o instanceof Flow && equals((Flow) o);
    }

    /**
     * Tells whether this flow is the root of the tree-structure of Flows in Brillien
     */
    protected boolean isRoot(){
        if( flow != null
            && (
                flow.equals( BrillienServices.SERVICES_FLOW ) || flow.equals( BrillienServices.SUPREME_FLOW )
            )
        )
            return true;
        return false;
    }

    @Override
    protected void report(String message, List<Object> params) {
        report( this, message, params );
    }

    protected void innerReport(Presence reporter, String message, List<Object> params){
        if( flow != null && !isRoot() )
            flow.report( reporter, message, params );
    }

    @Override
    public void report(Presence reporter, String message) {
        report( reporter, message, null );
    }

    public void report(Presence reporter, String message, List<Object> params) {
        if( message.equals( BrillienServices.REPORT_NEW_ASYNC_COMMUNICATION ) ){
            if( indispensable ){
                pendingAsyncCommunications++;
            } else if( !isRoot() ){
                flow.report( this, message, params );
            }
        } else if( message.equals( BrillienServices.REPORT_CLOSED_ASYNC_COMMUNICATION ) ){
            if( indispensable ){
                pendingAsyncCommunications--;
            } else if( !isRoot() ){
                flow.report( this, message, params );
            }
        } else{
            innerReport( reporter, message, params );
        }
    }

    @Override
    public <T> T getFlowResult(){
        return (T)result;
    }

    @Override
    public void setFlowResult( Object result ){
        this.result = result;
    }

    @Override
    public BrillienCommunication getActivationData(){
        return activationData;
    }

    @Override
    public void activate( BrillienCommunication c, Long timeout ) throws BrillienException{
        if( this.activationData != null )
            throw new BrillienException("Flow already started.");

        this.activationData = c;
        FlowServices.flowStarted( this, timeout, c );
    }

    @Override
    public void oppress( String message ) throws BrillienException{
        log( "Oppression in progress..." + getFlowID(), new InnerList<Object>( getFlowID(), getName(), message ) );

        addState( defaultErrorState );
        
        defaultErrorState.setName( message );

        passivateUnitMessaging();
    }

    @Override
    public void terminate() throws BrillienException{
        log( "Termination in progress..." + getFlowID(), new InnerList<Object>( getFlowID(), getName() ) );

        addState( defaultExitState );

        passivateUnitMessaging();
    }

    public boolean isSuspended(){
        return suspended;    
    }

    @Override
    public void suspend() throws BrillienException{
        log( "Marked as suspended:" + getFlowID(), new InnerList<Object>( getFlowID(), getName() ) );
        suspended = true;

        passivateUnitMessaging();        
    }

    @Override
    public void revivified() throws BrillienException{
        suspended = false;

        activateUnitMessaging();        
    }

    @Override
    protected State getDefaultState(){
        return defaultInitState;
    }
    
    @Override
    public void passivateUnitAll() throws BrillienException {
        super.passivateUnitAll();

        if( suspended )
            FlowServices.suspendFlow( this );

        else if( this.isUnitInExitState() )
            FlowServices.closeFlow( this );
        else
            FlowServices.rollbackFlow( this, "ERROR States:" + printUnitErrorStates( ) );
    }

    @Override
    protected void log( String message, List<Object> parameters ){
        if( message.endsWith(" report") ) return;

        super.log( message, parameters );
    }

    @Override
    protected void systemlog( String message, List<Object> parameters ){
        if( message.endsWith(" report") ) return;

        super.systemlog( message, parameters );
    }

    @Override
    public void retrieveUnit() throws BrillienException{
        if(
            this.equals( BrillienServices.SUPREME_FLOW ) ||
            this.equals( BrillienServices.SERVICES_FLOW )
        )
            exceptionLog( SuperFlow.class.getName(), "retrieveUnit", new BrillienException("Only non-system-level flows can be retrieved!") );
        else
            super.retrieveUnit();
    }

    @Override
    protected void innerChangeState( State from, State to ) throws BrillienException {
        super.innerChangeState( from, to );
        FlowServices.flowChagedState( this );
    }

    @Override
    public boolean isWaitingForResponse() {
        return super.isWaitingForResponse() || (pendingAsyncCommunications > 0);   
    }

    @Override
    protected Flow getFlowToReport(){
        return this;
    }

    @Override
    protected Connection getConnection( String dataSourceName, boolean autocommit, int isolation ) throws BrillienException {
        return TransactionServices.getConnection( this, dataSourceName, autocommit, isolation );
    }

    @Override
    public void reset(){
        super.reset();
        pendingAsyncCommunications  = 0;
        indispensable               = false;
        suspended                   = false;

        flowID                      = null;
        fXid                        = null;
        activationData              = null;
        result                      = null;
    }

}

